/**
 * Copyright © 2016-2022 The Thingsboard Authors
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.thingsboard.tools.service.device;

import com.alibaba.fastjson.JSONArray;
import io.netty.buffer.Unpooled;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.util.concurrent.Future;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Service;
import org.thingsboard.mqtt.MqttClient;
import org.thingsboard.server.common.data.Device;
import org.thingsboard.server.common.data.id.IdBased;
import org.thingsboard.tools.model.entity.AttributesDo;
import org.thingsboard.tools.model.entity.TelemetryDo;
import org.thingsboard.tools.service.mqtt.DeviceClient;
import org.thingsboard.tools.service.shared.BaseMqttAPITest;
import org.thingsboard.tools.service.shared.DefaultRestClientService;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

@Slf4j
@Service
@ConditionalOnProperty(prefix = "device", value = "api", havingValue = "MQTT")
public class MqttDeviceAPITest extends BaseMqttAPITest implements DeviceAPITest {


    @Override
    public void createDevices() throws Exception {

        //createDevices(true);
        List<Device> entities = createEntities(deviceStartIdx, deviceEndIdx);
        devices = Collections.synchronizedList(entities);
        //2:获取所有设备名称
        if (!devices.isEmpty()) {
            devicesNames = devices.stream().map(Device::getName).collect(Collectors.toList());
        } else {
            for (int i = deviceStartIdx; i < deviceEndIdx; i++) {
                devicesNames.add(getToken(false, i));
            }
        }
    }

    @Override
    public void removeDevices() throws Exception {
        removeEntities(devices.stream().map(IdBased::getId).collect(Collectors.toList()), "devices");
    }

    @Override
    public void runApiTests() throws InterruptedException {
        super.runApiTests(mqttClients.size());
    }

    @Override
    protected String getWarmUpTopic() {
        return "v1/devices/me/telemetry";
    }

    @Override
    protected byte[] getData(String deviceName) {
        TelemetryDo telemetryDo = new TelemetryDo();
        telemetryDo.setSn(deviceName);
        telemetryDo.setTs(System.currentTimeMillis()/1000);
        telemetryDo.setMo(mo);
        telemetryDo.setRs(rs);
        telemetryDo.setSi(si);
        telemetryDo.setTe(te);
        Object o = JSONArray.toJSON(telemetryDo);
        byte[] data = o.toString().getBytes(StandardCharsets.UTF_8);
        return data;
    }

    @Override
    protected void runApiTestIteration(int iteration, AtomicInteger totalSuccessPublishedCount, AtomicInteger totalFailedPublishedCount, CountDownLatch testDurationLatch) {
        runApiTestTelemetry(iteration, totalSuccessPublishedCount, totalFailedPublishedCount, testDurationLatch, false);
    }

    @Override
    protected String getTestTopic() {
        return telemetryTest ? "v1/devices/me/telemetry" : "v1/devices/me/attributes";
    }
    @Override
    protected String getAttributesTopic(){
        return "v1/devices/me/attributes";
    }
    @Override
    protected void logSuccessTestMessage(int iteration, DeviceClient client) {
        log.debug("[{}] Message was successfully published to device: {}", iteration, client.getDeviceName());
    }

    @Override
    protected void logFailureTestMessage(int iteration, DeviceClient client, Future<?> future) {
        log.error("[{}] Error while publishing message to device: {}", iteration, client.getDeviceName(), future.cause());
    }

    @Override
    public void connectDevices() throws InterruptedException {
        AtomicInteger totalConnectedCount = new AtomicInteger();
        List<String> pack = null;
        List<String> devicesNames;
        if (!devices.isEmpty()) {
            devicesNames = devices.stream().map(Device::getName).collect(Collectors.toList());
        } else {
            devicesNames = new ArrayList<>();
            for (int i = deviceStartIdx; i < deviceEndIdx; i++) {
                devicesNames.add(getToken(false, i));
            }
        }
        for (String device : devicesNames) {
            if (pack == null) {
                pack = new ArrayList<>(warmUpPackSize);
            }
            pack.add(device);
            if (pack.size() == warmUpPackSize) {
                connectDevices(pack, totalConnectedCount, false);
                Thread.sleep(1 + random.nextInt(100));
                pack = null;
            }
        }
        if (pack != null && !pack.isEmpty()) {
            connectDevices(pack, totalConnectedCount, false);
        }
        mapDevicesToDeviceClientConnections();
    }

    @Override
    public void generationX509() {

    }
    @Override
    public void warmUpGetWaySendDevices() throws Exception {
        //3:网关代发设备遥测数据
        AtomicInteger totalWarmedUpCount = new AtomicInteger();
        CountDownLatch packLatch = new CountDownLatch(devicesNames.size());
        int i =0;
        for(String devicesName:devicesNames){
            restClientService.getScheduler().submit(() -> {
                initGetwayKeyMqttClient.publish(getAttributesTopic(), Unpooled.wrappedBuffer(getData(devicesName)), MqttQoS.AT_MOST_ONCE)
                    .addListener(future -> {
                        if (future.isSuccess()) {
                            log.info("热身遥测数据发送成功 device: {}", devicesName);
                        } else {
                            log.error("热身遥测数据发送失败 device: {} error {}", devicesName,future.cause());
                        }
                        packLatch.countDown();
                        totalWarmedUpCount.getAndIncrement();
                    }
                );
            });
        }
        boolean succeeded = packLatch.await(10, TimeUnit.SECONDS);
        if (succeeded) {
            log.info("[{}] devices have been warmed up!", totalWarmedUpCount.get());
        } else {
            log.error("[{}] devices warmed up failed: {}!", totalWarmedUpCount.get(), packLatch.getCount());
        }
    }
    private void mapDevicesToDeviceClientConnections() {
        for (MqttClient mqttClient : mqttClients) {
            DeviceClient client = new DeviceClient();
            client.setMqttClient(mqttClient);
            client.setDeviceName(mqttClient.getClientConfig().getUsername());
            deviceClients.add(client);
        }
        log.info("Sorting device clients...");
        deviceClients.sort(Comparator.comparing(DeviceClient::getDeviceName));
        log.info("Shuffling device clients...");
        Collections.shuffle(deviceClients, random);
        log.info("Mapping devices to device client connections done");
    }

    /**
     *  重载创建设备方法
     * @param startIdx
     * @param endIdx
     * @return
     * @throws Exception
     */
    protected List<Device> createEntities(int startIdx, int endIdx) throws Exception{
        //1:拿到网关mqtt连接
        if (initGetwayKeyMqttClient == null){
            try {
                initGetwayKeyMqttClient = initClient(initGetwayKey);
            }catch (Exception e){
                log.error("初始化配置网关错误!! ",e);
            }
        }
        List<Device> result = Collections.synchronizedList(new ArrayList<>(1024 * 1024));
        int entityCount = endIdx - startIdx;
        log.info("Creating {} {}...", entityCount, ("devices"));
        for (int i = startIdx; i < endIdx; i++) {
            final int tokenNumber = i;
            String token = getToken(false, tokenNumber);
            Device entity = new Device();
            entity.setDeviceProfileId(deviceProfileManager.getByName(payloadType).getId());
            entity.setName(token);
            result.add(entity);
            AttributesDo attributesDo = new AttributesDo();
            attributesDo.setSn(initGetwayName);
            attributesDo.setNt("+");
            attributesDo.setNs(Arrays.asList(token));
            Object o = JSONArray.toJSON(attributesDo);
            byte[] data = o.toString().getBytes(StandardCharsets.UTF_8);
            initGetwayKeyMqttClient.publish(getAttributesTopic(), Unpooled.wrappedBuffer(data), MqttQoS.AT_MOST_ONCE)
                .addListener(future -> {
                    if (future.isSuccess()) {
                        log.info("创建设备成功 device: {}", token);
                    } else {
                        log.error("创建设备失败 device: {}", token);
                    }
                });
        }

        log.info("{} {} have been created successfully!", "devices");
        return result;
    }
}
